use std::{path::Path, sync::Arc, time::Duration};

use futures::{future::FutureExt, pin_mut, select};
use log::{error, info};
use tokio::{
    fs::{self, File},
    io::AsyncReadExt,
    sync::RwLock,
    task, time,
};

use crate::{Configure, Error, NodeId, Result};

mod carla_proxy;

mod manager_proxy;
use manager_proxy::{MapSettings, NodeCommand, ScenarioSettings, TaskSettings};

mod client_proxy;
use client_proxy::ConnectionId;

mod scenario_runner;

mod sensors_config;
use sensors_config::SensorsConfig;

mod core_state;

mod runtime_server;

mod coord_util;
use coord_util::*;

mod timer_util;

/// 开始服务.
pub async fn start(config: Configure, node_id: NodeId) -> Result<()> {
    Server::new(config, node_id).await?.run().await
}

/// 服务.
struct Server {
    manager_proxy: manager_proxy::Proxy,
    carla_proxy: carla_proxy::Proxy,
    carla_runtime_proxy: Option<carla_proxy::RuntimeProxy>,
    config: Configure,
    id: NodeId,
    scenario_proc: Option<scenario_runner::ProcessKiller>,
    core_state: Arc<RwLock<core_state::State>>,
}

impl Server {
    /// 创建状态机.
    async fn new(config: Configure, node_id: NodeId) -> Result<Self> {
        let manager_proxy =
            manager_proxy::Proxy::connect(config.broker_uri.as_str(), &node_id).await?;
        let (mut carla_proxy, mut carla_runtime_proxy) =
            carla_proxy::connect(config.carla_addr, config.carla_timeout_ms).await?;

        let snapshot = carla_runtime_proxy.get_snapshot().await?;

        let mut core_state = core_state::State::new(config.client_proxy_extern_uri.clone());
        core_state.set_map_name(carla_proxy.get_map_name().await?);
        core_state.set_running_time(snapshot.running_time);

        Ok(Self {
            manager_proxy,
            carla_proxy,
            carla_runtime_proxy: Some(carla_runtime_proxy),
            config: config.clone(),
            id: node_id,
            scenario_proc: None,
            core_state: Arc::new(RwLock::new(core_state)),
        })
    }

    /// 单步运行.
    ///
    /// 外部消息或定时器触发更新一次状态,未出错则可继续调用.
    async fn run(&mut self) -> Result<()> {
        let core_state = self.core_state.clone();

        let (mut client_proxy_sender, client_proxy_reciever) =
            client_proxy::bind(self.config.client_proxy_uri.as_str()).await?;
        let (client_proxy_reciever, client_proxy_runtime_reciever) = client_proxy_reciever.split();

        let runtime_server = runtime_server::Server::new(
            client_proxy_runtime_reciever,
            client_proxy_sender.clone(),
            self.carla_runtime_proxy.take().unwrap(),
            core_state.clone(),
        );

        let cmd_fut = self.manager_proxy.get_command_watch().get_command().fuse();
        let req_fut = client_proxy_reciever.recv().fuse();
        let timer_fut = time::sleep(Duration::from_millis(0)).fuse();
        let wait_runtime_fut = runtime_server.wait().fuse();
        pin_mut!(cmd_fut, req_fut, timer_fut, wait_runtime_fut);

        loop {
            select! {
                // 响应管理端下发的控制命令
                cmd = cmd_fut => {
                    let (command_watch, NodeCommand {
                        id,
                        server_type,
                        task,
                        scenario,
                        map,
                    }) = cmd?;

                    {
                        let mut core_state = core_state.write().await;

                        if !self.config.command_id_filter || !core_state.has_command(id.as_str()) {
                            self.unset_scenario(&mut core_state).await?;
                            if let Some(scenario) = scenario {
                                self.set_scenario(scenario, None, &mut core_state).await?;
                            } else if let Some(map) = map {
                                self.set_map(&map, &mut core_state).await?;
                            }

                            if let Some(task) = task {
                                self.set_task(&task, &mut core_state).await?;
                            } else {
                                self.unset_task(&mut core_state).await?;
                            }

                            core_state.set_command(id);
                            core_state.set_server_type(server_type);
                        }
                    }

                    cmd_fut.set(command_watch.get_command().fuse());
                }

                // 响应客户端的请求
                req = req_fut => {
                    let (client_proxy_reciever, timeout_conn_ids, req) = req?;

                    {
                        self.delete_timeout_vehicles(&timeout_conn_ids[..], & core_state).await?;

                        if let Some((conn_id, req)) = req {
                            match req {
                                client_proxy::UpMessage::CreateVehicle(msg) => {
                                    let ret = self.create_vehicle(&conn_id, msg.client_id.as_str(),
                                        msg.vehicle_type.as_str(), msg.position_xyz,
                                        msg.orientation_wxyz, &core_state).await?;

                                    if let Err(e) = client_proxy_sender.send(conn_id, client_proxy::DownMessage::response(ret)).await {
                                        error!("{}", e);
                                    }
                                }

                                client_proxy::UpMessage::GetVehicle(msg) => {
                                    let ret = self.get_vehicle(&conn_id, msg.client_id.as_str(),
                                        msg.token.as_str(), &core_state)
                                        .await?;

                                    if let Err(e) = client_proxy_sender.send(conn_id, client_proxy::DownMessage::response(ret)).await {
                                        error!("{}", e);
                                    }
                                }

                                client_proxy::UpMessage::SetVehicleTransform(msg) => {
                                    let ret = self.set_vehicle_transform(&conn_id,
                                    msg.position_xyz, msg.orientation_wxyz, &core_state)
                                        .await?;

                                    if let Err(e) = client_proxy_sender.send(conn_id, client_proxy::DownMessage::response(ret)).await {
                                        error!("{}", e);
                                    }
                                }

                                client_proxy::UpMessage::DetachVehicle => {
                                    let ret = self.delete_vehicle(&conn_id, &core_state).await?;

                                    if let Err(e) = client_proxy_sender.send(conn_id, client_proxy::DownMessage::response(ret)).await {
                                        error!("{}", e);
                                    }
                                }

                                _ => ()
                            }
                        }
                    }

                    req_fut.set(client_proxy_reciever.recv().fuse());
                }

                // 响应定时器
                _ = timer_fut => {
                    {
                        let core_state = core_state.read().await;

                        self.publish_node_status(&core_state).await?;
                    }

                    timer_fut.set(time::sleep(Duration::from_millis(
                        self.config.publish_status_period_ms as u64)).fuse());
                }

                // runtime_server结束时退出
                wait_runtime = wait_runtime_fut => {
                    if let Err(e) = &wait_runtime {
                        error!("{}", e);
                    }

                    return wait_runtime;
                }
            }
        }
    }

    /// 发布节点状态到消息总线.
    async fn publish_node_status(&mut self, core_state: &core_state::State) -> Result<()> {
        if let Err(e) = self
            .manager_proxy
            .publish_node_status(&self.id, &core_state.to_node_status())
            .await
        {
            error!("{}", e);
        }

        Ok(())
    }

    /// 加载场景.
    async fn set_scenario(
        &mut self,
        scenario: ScenarioSettings,
        map: Option<MapSettings>,
        core_state: &mut core_state::State,
    ) -> Result<()> {
        let reload = scenario.force_reload || core_state.has_scenario(scenario.name.as_str());

        let mut load_map = false;

        if reload {
            if scenario.is_package {
                match self.manager_proxy.download(scenario.url.as_str()).await {
                    Ok(content) => {
                        let r = extract_zip_files(&self.config.tmp_scenario_package_dir, content)
                            .await?;

                        if !r {
                            error!("Extract scenario package failed");
                            return Ok(());
                        }
                    }
                    Err(e) => {
                        error!("{}", e);
                        return Ok(());
                    }
                }
            } else if scenario.is_openscenario {
                if let Some(map) = &map {
                    match self.manager_proxy.download(map.url.as_str()).await {
                        Ok(content) => {
                            write_file(&self.config.tmp_map_path, &content[..]).await?;

                            load_map = true;

                            match self.manager_proxy.download(scenario.url.as_str()).await {
                                Ok(content) => {
                                    write_file(&self.config.tmp_openscenario_path, &content[..])
                                        .await?;
                                }
                                Err(e) => {
                                    error!("{}", e);
                                    return Ok(());
                                }
                            }
                        }
                        Err(e) => {
                            error!("{}", e);
                            return Ok(());
                        }
                    }
                } else {
                    error!("No map");
                    return Ok(());
                }
            } else {
                match self.manager_proxy.download(scenario.url.as_str()).await {
                    Ok(content) => {
                        write_file(&self.config.tmp_scenario_cfg_path, &content[..]).await?;
                    }
                    Err(e) => {
                        error!("{}", e);
                        return Ok(());
                    }
                }
            }
        }

        let (proc, proc_killer) = scenario_runner::Process::new(
            &self.config,
            scenario.name.as_str(),
            scenario.is_openscenario,
            scenario.is_package,
        )
        .await?;

        self.scenario_proc = Some(proc_killer);

        core_state.set_map_name("".into());
        if load_map {
            if let Some(map) = map {
                core_state.set_map_name(map.name);
            }
        }
        let scenario_id = core_state.set_scenario(scenario.name);
        core_state.set_running_time(0);

        let core_state = self.core_state.clone();
        tokio::spawn(async move {
            match proc.wait().await {
                Ok(scenario_result) => {
                    if scenario_result.success {
                        info!("Run scenario success");
                    } else {
                        info!("Run scenario failed");
                    }

                    core_state.write().await.set_scenario_stop(
                        Some(scenario_id),
                        true,
                        scenario_result.success,
                    );
                }

                Err(e) => {
                    error!("{}", e);
                    core_state
                        .write()
                        .await
                        .set_scenario_stop(Some(scenario_id), false, false);
                }
            }
        });

        info!("Start scenario");

        Ok(())
    }

    /// 停止当前场景运行.
    async fn unset_scenario(&mut self, core_state: &mut core_state::State) -> Result<()> {
        if let Some(proc) = self.scenario_proc.take() {
            proc.kill().await;
        }

        core_state.set_scenario_stop(None, false, false);

        // 删除所有车辆
        core_state.delete_all_players();
        self.carla_proxy.destroy_all_vehicles().await?;

        Ok(())
    }

    /// 加载地图.
    async fn set_map(
        &mut self,
        map: &MapSettings,
        core_state: &mut core_state::State,
    ) -> Result<()> {
        let reload = map.force_reload || core_state.has_map(map.name.as_str());
        if reload {
            match self.manager_proxy.download(map.url.as_str()).await {
                Ok(content) => match String::from_utf8(content) {
                    Ok(content) => match self.carla_proxy.load_map(content.clone()).await {
                        Ok(_) => {
                            write_file(&self.config.tmp_map_path, content.as_bytes()).await?;

                            core_state.set_map_name(map.name.clone());
                            core_state.set_running_time(0);
                        }

                        Err(e) => {
                            error!("{}", e);
                        }
                    },

                    Err(e) => {
                        error!("{}", e);
                    }
                },

                Err(e) => {
                    error!("{}", e);
                }
            }
        }

        Ok(())
    }

    /// 创建车辆.
    async fn create_vehicle(
        &mut self,
        conn_id: &ConnectionId,
        client_id: &str,
        vehicle_type: &str,
        position_xyz: (f64, f64, f64),
        orientation_wxyz: (f64, f64, f64, f64),
        core_state: &RwLock<core_state::State>,
    ) -> Result<bool> {
        {
            let mut core_state = core_state.write().await;

            if let Some((_, client_created, vehicle_id, _)) = core_state.get_player_info(conn_id) {
                self.carla_proxy
                    .destroy_vehicle(vehicle_id.as_str(), client_created)
                    .await?;

                core_state.delete_player(conn_id);
            }
        }

        match self
            .carla_proxy
            .spawn_vehicle(
                vehicle_type,
                n2c_location(position_xyz),
                n2c_rotation(orientation_wxyz),
            )
            .await
        {
            Ok(vehicle_id) => {
                if self
                    .enable_vehicle_carsim(vehicle_id.as_str(), vehicle_type)
                    .await?
                {
                    // TODO:
                    let sensors_config = SensorsConfig::default();

                    if self
                        .enable_vehicle_sensors(vehicle_id.as_str(), vehicle_type)
                        .await?
                    {
                        if core_state
                            .write()
                            .await
                            .add_player(
                                conn_id.clone(),
                                client_id.to_string(),
                                true,
                                vehicle_id.clone(),
                                vehicle_type.to_string(),
                                sensors_config,
                            )
                            .is_err()
                        {
                            error!("Logic error");
                        }

                        info!("Create vehicle {}, type {}", vehicle_id, vehicle_type);

                        return Ok(true);
                    }
                }

                self.carla_proxy
                    .destroy_vehicle(vehicle_id.as_str(), true)
                    .await?;

                Ok(false)
            }

            Err(e) => {
                error!("{}", e);

                Ok(false)
            }
        }
    }

    /// 获取场景中创建好的车辆.
    async fn get_vehicle(
        &mut self,
        conn_id: &ConnectionId,
        client_id: &str,
        role_name: &str,
        core_state: &RwLock<core_state::State>,
    ) -> Result<bool> {
        {
            let mut core_state = core_state.write().await;

            if let Some((_, client_created, vehicle_id, _)) = core_state.get_player_info(conn_id) {
                self.carla_proxy
                    .destroy_vehicle(vehicle_id.as_str(), client_created)
                    .await?;

                core_state.delete_player(conn_id);
            }
        }

        match self.carla_proxy.get_vehicle_by_role(role_name).await {
            Ok((vehicle_id, vehicle_type)) => {
                if self
                    .enable_vehicle_carsim(vehicle_id.as_str(), vehicle_type.as_str())
                    .await?
                {
                    // TODO:
                    let sensors_config = SensorsConfig::default_with_camera();

                    if self
                        .enable_vehicle_sensors(vehicle_id.as_str(), vehicle_type.as_str())
                        .await?
                    {
                        if core_state
                            .write()
                            .await
                            .add_player(
                                conn_id.clone(),
                                client_id.to_string(),
                                false,
                                vehicle_id.clone(),
                                vehicle_type.clone(),
                                sensors_config,
                            )
                            .is_err()
                        {
                            error!("Logic error");
                        }

                        info!("Attach vehicle {}, type {}", vehicle_id, vehicle_type);

                        return Ok(true);
                    }
                }

                Ok(false)
            }

            Err(e) => {
                error!("{}", e);

                Ok(false)
            }
        }
    }

    /// 启用车辆的carsim支持.
    async fn enable_vehicle_carsim(
        &mut self,
        vehicle_id: &str,
        vehicle_type: &str,
    ) -> Result<bool> {
        if let Some(carsim_cfg) = self.get_carsim_cfg(vehicle_type).await {
            match self
                .carla_proxy
                .enable_carsim(vehicle_id, carsim_cfg.as_str())
                .await
            {
                Ok(_) => {
                    info!("Enable carsim for {}", vehicle_id);
                    Ok(true)
                }

                Err(e) => {
                    error!("{}", e);
                    Ok(false)
                }
            }
        } else {
            Ok(true)
        }
    }

    /// 启用车辆传感器.
    async fn enable_vehicle_sensors(
        &mut self,
        vehicle_id: &str,
        vehicle_type: &str,
    ) -> Result<bool> {
        let sensors_config = if let Some(config) = self.get_sensors_cfg(vehicle_type).await {
            SensorsConfig::parser(&config[..])?
        } else {
            SensorsConfig::default()
        };

        match self
            .carla_proxy
            .enable_base_sensors(
                vehicle_id,
                n2c_location(sensors_config.gnss.position_xyz),
                n2c_rotation(sensors_config.gnss.orientation_wxyz),
                sensors_config.gnss.period as f32,
            )
            .await
        {
            Ok(_) => {
                info!("Enable base sensors for {}", vehicle_id);

                for (camera_id, config) in &sensors_config.cameras {
                    match self
                        .carla_proxy
                        .enable_camera(
                            vehicle_id,
                            camera_id.as_str(),
                            n2c_location(config.position_xyz),
                            n2c_rotation(config.orientation_wxyz),
                            config.fov as f32,
                            config.image_width,
                            config.image_height,
                            config.iso as f32,
                            config.gamma as f32,
                            config.shutter_speed as f32,
                            config.period as f32,
                        )
                        .await
                    {
                        Ok(_) => {
                            info!("Enable camera {} for {}", camera_id, vehicle_id);
                        }

                        Err(e) => {
                            error!("{}", e);
                            return Ok(false);
                        }
                    }
                }

                Ok(true)
            }

            Err(e) => {
                error!("{}", e);
                Ok(false)
            }
        }
    }

    /// 设置车辆位置.
    async fn set_vehicle_transform(
        &mut self,
        conn_id: &ConnectionId,
        position_xyz: (f64, f64, f64),
        orientation_wxyz: (f64, f64, f64, f64),
        core_state: &RwLock<core_state::State>,
    ) -> Result<bool> {
        let player_info = core_state.read().await.get_player_info(conn_id);
        if let Some((_, _, vehicle_id, _)) = player_info {
            Ok(self
                .carla_proxy
                .set_transform(
                    vehicle_id.as_str(),
                    n2c_location(position_xyz),
                    n2c_rotation(orientation_wxyz),
                )
                .await
                .is_ok())
        } else {
            Ok(false)
        }
    }

    /// 删除车辆.
    async fn delete_vehicle(
        &mut self,
        conn_id: &ConnectionId,
        core_state: &RwLock<core_state::State>,
    ) -> Result<bool> {
        let player_info = core_state.read().await.get_player_info(conn_id);
        if let Some((client_id, client_created, vehicle_id, _)) = player_info {
            self.carla_proxy
                .destroy_vehicle(vehicle_id.as_str(), client_created)
                .await?;

            info!("Delete vehicle {}", vehicle_id);

            core_state.write().await.delete_player(conn_id);

            info!("Client {} disconnected", client_id);
        }

        Ok(true)
    }

    /// 删除过期车辆.
    async fn delete_timeout_vehicles(
        &mut self,
        conn_ids: &[ConnectionId],
        core_state: &RwLock<core_state::State>,
    ) -> Result<()> {
        if !conn_ids.is_empty() {
            let vehicles = {
                let core_state = core_state.read().await;

                conn_ids
                    .iter()
                    .filter_map(|conn_id| {
                        core_state
                            .get_player_info(conn_id)
                            .map(|(_, client_created, vehicle_id, _)| (vehicle_id, client_created))
                    })
                    .collect::<Vec<_>>()
            };

            for (vehicle_id, client_created) in vehicles {
                self.carla_proxy
                    .destroy_vehicle(vehicle_id.as_str(), client_created)
                    .await?;

                info!("Destroy vehicle {}", vehicle_id);
            }

            {
                let mut core_state = core_state.write().await;

                for conn_id in conn_ids {
                    if let Some((client_id, _, _, _)) = core_state.get_player_info(conn_id) {
                        info!("Client {} disconnected", client_id);
                    }

                    core_state.delete_player(conn_id);
                }
            }
        }

        Ok(())
    }

    /// 根据车型获取对应的配置.
    async fn get_carsim_cfg(&self, vehicle_type: &str) -> Option<String> {
        let file_path = self.config.carsim_cfg_path.join(vehicle_type);
        if let Ok(mut file) = File::open(&file_path).await {
            let mut cfg = String::new();
            file.read_to_string(&mut cfg).await.map(|_| cfg).ok()
        } else {
            None
        }
    }

    /// 根据车型获取对应的传感器配置.
    async fn get_sensors_cfg(&self, vehicle_type: &str) -> Option<Vec<u8>> {
        let file_path = self
            .config
            .sensors_cfg_path
            .join(format!("{}.json", vehicle_type));
        if let Ok(mut file) = File::open(&file_path).await {
            let mut buf = Vec::new();
            if file.read_to_end(&mut buf).await.is_ok() {
                return Some(buf);
            }
        }

        None
    }

    /// 加载任务.
    async fn set_task(
        &mut self,
        task: &TaskSettings,
        core_state: &mut core_state::State,
    ) -> Result<()> {
        core_state.set_task(task.id.clone(), task.token.clone());

        Ok(())
    }

    /// 删除任务.
    async fn unset_task(&mut self, core_state: &mut core_state::State) -> Result<()> {
        core_state.unset_task();

        Ok(())
    }
}

/// 创建全路径并写入文件.
async fn write_file(path: impl AsRef<Path>, contents: &[u8]) -> Result<()> {
    if let Some(dir) = path.as_ref().parent() {
        fs::create_dir_all(dir).await.map_err(Error::from)?;
    }

    fs::write(path.as_ref(), contents)
        .await
        .map_err(Error::from)?;

    Ok(())
}

/// 解压场景文件.
async fn extract_zip_files(dir: impl AsRef<Path>, contents: Vec<u8>) -> Result<bool> {
    fs::remove_dir_all(dir.as_ref()).await.ok();
    fs::create_dir_all(dir.as_ref())
        .await
        .map_err(Error::from)?;

    if let Ok(mut zip) = zip::ZipArchive::new(std::io::Cursor::new(contents)) {
        let dir = dir.as_ref().to_path_buf();
        task::spawn_blocking(move || -> Result<bool> {
            zip.extract(dir).map_or(Ok(false), |_| Ok(true))
        })
        .await
        .map_err(Error::from)?
    } else {
        Ok(false)
    }
}
